package com.yanxu;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;

import java.util.Random;

/**
 * @author 折戟沉沙铁未销
 * @version V1.0
 * @date 2025-07-13-2025
 * @Description: api 学习
 */
public class Api05_Source_Custom {
    public static void main(String[] args) throws Exception {
        //获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //源 -算子

        //从自定义 source 中接收数据
        DataStreamSource<Integer> streamSource = env.addSource(new CustomSourceParalle()).setParallelism(2);
        streamSource.print("custom paralle source >>>");


        //对环境进行执行。
        env.execute();
    }
}

@Slf4j
class CustomSourceParalle implements ParallelSourceFunction<Integer> {

    // 声明一个布尔型变量，用于控制数据源是否继续运行
    private boolean running = true;
    // 声明 random 对象
    private Random random = new Random();


    /**
     * 数据源的核心方法，用于产生数据
     *
     * @param ctx 上下文对象，用于收集产生的数据
     * @throws Exception 可能抛出的异常
     */
    @Override
    public void run(SourceContext<Integer> ctx) throws Exception {
        while (running) {
            int rad = random.nextInt();
            log.info("{}",rad);
            ctx.collect(rad);
        }

    }

    /**
     * 取消数据生成的方法，设置running为false停止数据源
     */
    @Override
    public void cancel() {
        running = false;
    }
}